package com.atguigu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.DateFormatUtil;
import com.atguigu.utils.KafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

//数据流:web/app -> logfile -> flume -> kafka(ODS) -> FlinkApp -> Kafka(DWD)
//程 序：Mock -> logfile -> f1.sh -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK)
/*
主流+侧输出流
流量域事务事实表:
	流量域页面浏览事务事实表
	流量域启动事务事实表
	流量域动作事务事实表
	流量域曝光事务事实表
	流量域错误事务事实表

 */
//todo 1.获取执行环境
//todo 2.读取kafka topic_log主题数据创建流
//todo 3.过滤掉非json格式数据并转为json对象
//todo 4.按照mid分组
//todo 5.使用状态编程的方式实现新老用户校验
//todo 6.使用侧输出流分流5个流 将页面日志从主流输出，其他的从侧输出流输出
//todo 7.获取主流和侧输出流的数据并写到kafka各自的主题
//todo 8.启动任务
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //todo 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //todo 设置状态后端

        //todo 2.读取kafka topic_log主题数据创建流
        DataStreamSource<String> kafkaDS = env.addSource(KafkaUtil.getFlinkKafkaConsumer("topic_log", "baselogapp2_220828"));

        //todo 3.过滤掉非json格式数据并转为json对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    try {
                        JSONObject jsonObject = JSON.parseObject(value);
                        out.collect(jsonObject);
                    } catch (JSONException e) {
                        System.out.println("过滤了脏数据（非法json）：" + value);
                    }
                }
            }
        });
 /*
        value:日志数据，经过上述操作变成了String类型
        {
    "actions":[
        {
            "action_id":"favor_add",
            "item":"33",
            "item_type":"sku_id",
            "ts":1651303989298
        },
        {
            "action_id":"cart_add",
            "item":"33",
            "item_type":"sku_id",
            "ts":1651303993596
        }
    ],
    "common":{
        "ar":"440000",
        "ba":"iPhone",
        "ch":"Appstore",
        "is_new":"0",
        "md":"iPhone X",
        "mid":"mid_3442864",
        "os":"iOS 12.4.1",
        "uid":"928",
        "vc":"v2.1.134"
    },
    "displays":[
        {
            "display_type":"promotion",
            "item":"19",
            "item_type":"sku_id",
            "order":1,
            "pos_id":3
        },
        {
            "display_type":"query",
            "item":"27",
            "item_type":"sku_id",
            "order":2,
            "pos_id":4
        }
    ],
    "page":{
        "during_time":17195,
        "item":"33",
        "item_type":"sku_id",
        "last_page_id":"good_list",
        "page_id":"good_detail",
        "source_type":"promotion"
    },
    "ts":1651303985000
}
         */
        //todo 4.按照mid分组
        KeyedStream<JSONObject, String> keyByMidDS = jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("mid"));

        //todo 5.使用状态编程的方式实现新老用户校验
        SingleOutputStreamOperator<JSONObject> jsonObjWithNewFlagDS = keyByMidDS.map(new RichMapFunction<JSONObject, JSONObject>() {
            //            状态存放该访客（mid）上一次登录的日期
            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //状态初始化
                valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastDt-state", String.class));

            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {
                //获取状态
                String lastDt = valueState.value();
                //获取is_new
                String isNew = value.getJSONObject("common").getString("is_new");
                //获取今天的日期
                String curDt = DateFormatUtil.toDate(value.getLong("ts"));


                //新老用户校验
                if ("1".equals(isNew)) {//前端发来1不一定是新用户
                    if (lastDt == null || lastDt.equals(curDt)) {
                        //lastDtState == null说明是该设备的第一条数据，lastDtState.equals(curDt)说明是今天，并且第一次登录在今天，今天多次登录都表示是新用户，都表示是新用户
                        valueState.update(curDt);
                    } else {//lastDtState != null && 登录日期不等于状态日期，都表示是老用户，要把value里的is_new改为0再返回
                        value.getJSONObject("common").put("is_new", "0");
                        //状态日期就不用更新了（如果更新为今天，今天卸载，又重新登录，则前端发来1，登录日期=状态日期，走上面的方法会被认为是新用户，是错误的）
                    }

                } else {//isNEW=0 前端发来0就一定是老用户
                    if (lastDt == null) {//老用户并且状态为null，说明实时上线之前就登录了
                        valueState.update("1970-01-01");

                    }

                }

                return value;
            }
        });

        //todo 6.使用侧输出流分流5个流 将页面日志从主流输出，其他的从侧输出流输出
//        注意！！！ {}不能丢
        OutputTag<JSONObject> startTag = new OutputTag<JSONObject>("start") {
        };//the type of elements in the side-output stream.
        OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>("display") {
        };
        OutputTag<JSONObject> actionTag = new OutputTag<JSONObject>("action") {
        };
        OutputTag<JSONObject> errorTag = new OutputTag<JSONObject>("error") {
        };

        //侧输出流只能用process
        SingleOutputStreamOperator<JSONObject> pageDS = jsonObjWithNewFlagDS.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
                //1.处理error错误日志
                String err = value.getString("err");
                if (err != null) {
                    //说明有错误信息
                    ctx.output(errorTag, value);//err保留了整个原始json
                    //将错误信息移除，后面要过滤出start等其他日志
                    value.remove("err");
                }

                //2.处理start启动日志(来的日志不是启动日志，就是页面日志)
                String start = value.getString("start");
                if (start != null) {
                    ctx.output(startTag, value);//start里面不会有err信息
                } else {
                    //3.处理页面日志(来的日志不是启动日志，就是页面日志)
                    //3.1获取曝光日志,遍历displays里的每一个元素，输出到侧输出流，并添加ts和page_id字段
                    Long ts = value.getLong("ts");
                    String pageId = value.getJSONObject("page").getString("page_id");
                    JSONArray displays = value.getJSONArray("displays");
                    JSONArray actions = value.getJSONArray("actions");

                    if (displays != null && displays.size() > 0) {//displays.size()>0：不为[]，此时的页面日志只有page、displays和actions信息
                        for (Object display : displays) {
                            JSONObject jsonObject = JSONObject.parseObject(display.toString());
                            jsonObject.put("ts", ts);
                            jsonObject.put("page_id", pageId);
                            ctx.output(displayTag, jsonObject);//含display、ts、page_id
                            value.remove("displays");
                        }
                    } else if (actions != null && actions.size() > 0) { //此时的页面日志只有page和actions信息，没有错误信息，和displays信息
                        for (Object action : actions) {
                            JSONObject jsonObject = JSONObject.parseObject(action.toString());
                            jsonObject.put("ts", ts);
                            jsonObject.put("page_id", pageId);
                            ctx.output(actionTag, jsonObject);//含action、common、ts、page_id
                            value.remove("actions");
                        }
                    }

                    out.collect(value);//主流：页面日志，含page、common、ts、page_id


                }
            }
        });
        //todo 7.获取侧输出流的数据并写到kafka
        DataStream<String> startDS = pageDS.getSideOutput(startTag).map(JSON::toString);
        DataStream<String> errorDS = pageDS.getSideOutput(errorTag).map(JSON::toString);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag).map(JSON::toString);
        DataStream<String> actionDS = pageDS.getSideOutput(actionTag).map(JSON::toString);
        SingleOutputStreamOperator<String> pageDS1 = pageDS.map(JSON::toString);

        startDS.print("startDS>>>>");
        errorDS.print("errorDS>>>>");
        displayDS.print("displayDS>>>>");
        actionDS.print("actionDS>>>>");
        pageDS.print("pageDS>>>>");

        startDS.addSink(KafkaUtil.getFlinkKafkaProducer("start_topic"));
        errorDS.addSink(KafkaUtil.getFlinkKafkaProducer("error_topic"));
        displayDS.addSink(KafkaUtil.getFlinkKafkaProducer("display_topic"));
        actionDS.addSink(KafkaUtil.getFlinkKafkaProducer("action_topic"));
        pageDS1.addSink(KafkaUtil.getFlinkKafkaProducer("page_topic"));



        //todo 8.启动任务
        env.execute("");


    }
}
